package com.pw.study.flink.watermark

import com.pw.study.flink.source.{MyWordSource, WordBean, WordSource}
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow

/**
 * @Desc: TODO
 * @author: future
 * @since: 2022/5/15-16:06
 */
object MyWatermark {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(2)
    val ds= env.addSource(new WordSource())
    val dsW: DataStream[WordBean] = ds.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[WordBean](Time.seconds(3)) {
      override def extractTimestamp(bean: WordBean): Long = {
        bean.ts
      }
    })
    val dsWindow: WindowedStream[WordBean, String, TimeWindow] = dsW.keyBy(_.word).window(TumblingEventTimeWindows.of(Time.seconds(1)))
    val dsResult: DataStream[WordBean] = dsWindow.reduce((x, y) => {
      val one: Int = x.one + y.one
      println("时间。。。。。。。。。。。。。。。。。。。。。。。。")
      val bean: WordBean = WordBean(x.word, one, Math.max(x.ts, y.ts))
      println("bean:"+bean)
      bean
    })

    dsW.print("water")
    //ds.print()
    dsResult.print("result:")


    env.execute("watermark")
  }
}
